/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: cbus_client.c
 *
 * Purpose: implementation client side for cbus.
 *
 * Developer:
 *   wen.gu , 2022-02-25
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/
#include "cbus/bus/cbus_client.h"

#include <string.h>

#include "cbus/bus/cbus_endpoint.h"
#include "cbus/bus/cbus_message_manager.h"
#include "cbus/bus/cbus_endpoint_ops.h"
#include "cbus/core/cbus_map.h"
#include "cbus/core/cbus_mutex.h"

#include "cbus/core/cbus_log.h"
/******************************************************************************
 **    MACROS
 ******************************************************************************/

#define REQ_MAP_CREATE(req_map, msg_type) do{                                         \
    req_map[msg_type] = cbus_map_create(CMV_TYPE_Pointer, CBUS_MAP_DEFAULT_CAPACITY); \
    if (!req_map[msg_type]){                                                          \
    LOGE("alloc map instance for response msg type(%d) failed\n", msg_type);          \
    return G_ErrInsufficientResources;}}while(0)

#define REQ_CHECK_AND_RETURN(check_value, check_ok, return_value, free_func) do{ \
    if (check_value != check_ok){free_func; return return_value;}}while(0)
/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/

//typedef void (*cbus_client_response_handler)(GPTR opaque, cbus_message_id message_id, cbus_session_id session_id, GU08 ret_code, cbus_buffer_t response_data);

typedef cbus_map_t cbus_client_response_t;

typedef struct _client_response_info
{
    cbus_mutex_t lock;
    cbus_condition_variable_t cond;
    cbus_message_t response_message;

    GPTR opaque;
    cbus_client_on_response_func on_response;
}client_response_info_t;

typedef struct _client_event_info
{
    GPTR opaque;
    cbus_client_on_event_func on_event;
}client_event_info_t;

struct _cbus_client 
{
    GU08 interface_version;  /** the version of service provided interface */
    GU16 next_session_id; /** todo, refine me, is need calculate session id for every message_type and every message_id independence */

    cbus_mutex_t lock; /** a lock for op response map (add, remove, find/get, set/update)*/

    cbus_map_t event_map;
    cbus_map_t property_event_map;


    cbus_endpoint_t endpoint;

    /** like: map<method_id, session_map>
     *  session_map like:  map<session_id, session_info*>
     */
    cbus_client_response_t response_map[CBUS_MSG_TYPE_MAX];  
};

/******************************************************************************
 **    inner FUNCTION DEFINITIONS
 ******************************************************************************/

static void client_map_free(void* ptr)
{
    cbus_map_destroy((cbus_map_t)ptr);
}

static GErrc response_map_init(cbus_client_t client)
{
    REQ_MAP_CREATE(client->response_map, CBUS_MSG_TYPE_Method_Response);
    REQ_MAP_CREATE(client->response_map, CBUS_MSG_TYPE_Event_Subscribe_Ack);
    REQ_MAP_CREATE(client->response_map, CBUS_MSG_TYPE_Event_Stop_Subscribe_Ack);
    REQ_MAP_CREATE(client->response_map, CBUS_MSG_TYPE_Property_Get_Ack);
    REQ_MAP_CREATE(client->response_map, CBUS_MSG_TYPE_Property_Set_Ack);
    REQ_MAP_CREATE(client->response_map, CBUS_MSG_TYPE_Property_Subscribe_Ack);
    REQ_MAP_CREATE(client->response_map, CBUS_MSG_TYPE_Property_Stop_Subscribe_Ack);

    return G_OK;
}

static cbus_session_id cbus_client_get_session_id(cbus_client_t client)
{
    return client->next_session_id++; /** todo refine me? mutex or atomic op */
}

/** the response op for client */
static void client_response_info_destroy(void* ptr)
{
    if (ptr)
    {
        client_response_info_t* info = (client_response_info_t*)ptr;

        if (info->cond)
        {
            if (info->response_message)
            {
                cbus_message_free(info->response_message);
                info->response_message = NULL;
            }

            cbus_cv_notify_all(info->cond);
            cbus_cv_destroy(info->cond);
            info->cond = NULL;
        }

        if (info->lock)
        {
            cbus_mutex_destroy(info->lock);
            info->lock = NULL;
        }

        free(ptr);
    }
}

static client_response_info_t* client_response_info_create(cbus_client_on_response_func on_response, GPTR opaque)
{

    client_response_info_t* info = (client_response_info_t*)malloc(sizeof(client_response_info_t));

    if (info)
    {
        memset(info, 0, sizeof(client_response_info_t));

        info->lock = cbus_mutex_create();

        if (!info->lock)
        {
            client_response_info_destroy(info);
            return NULL;
        }

        info->cond = cbus_cv_create();

        if (!info->cond)
        {
            client_response_info_destroy(info);
            return NULL;
        }

        info->on_response = on_response;
        info->opaque = opaque;
    }

    return info;
}

static client_response_info_t* client_response_get_info(cbus_client_t client, cbus_client_response_t response, cbus_message_id message_id, cbus_session_id session_id)
{
    cbus_map_t session_map = NULL;
    if (cbus_map_find_pointer(response, message_id, (const void**)&session_map) == GTRUE)
    {
        client_response_info_t* info = NULL;
        cbus_map_find_pointer(session_map, session_id, (const void**)&info);
        return info;
    }

    return NULL;
}

static GErrc msg_response_set_data(cbus_client_t client, cbus_client_response_t response, cbus_message_t response_message)
{
    cbus_map_t session_map = NULL;
    cbus_mutex_lock(client->lock);

    if (cbus_map_find_pointer(response, cbus_message_get_message_id(response_message), (const void**)&session_map) == GFALSE)
    {
        cbus_mutex_unlock(client->lock);
        return G_ErrNotFound;
    }

    client_response_info_t* info = NULL;
    if (cbus_map_find_pointer(session_map, cbus_message_get_session_id(response_message), (const void**)&info) == GFALSE)
    {
        cbus_mutex_unlock(client->lock);
        return G_ErrNotFound;
    }
    cbus_mutex_unlock(client->lock);

    cbus_mutex_lock(info->lock);
    if (info->on_response) /** for async request */
    {
        cbus_message_set_opaque(response_message, info->opaque);
        info->on_response(response_message);
        cbus_mutex_unlock(info->lock);
        /** after response callback, then need remove session info from map */
        cbus_mutex_lock(client->lock);
        cbus_map_remove(session_map, cbus_message_get_session_id(response_message));
        cbus_mutex_unlock(client->lock);        
    }
    else
    { /** for sync request */        
        info->response_message = cbus_message_reference(response_message);
        //todo is need set opaque at here?
        cbus_cv_notify_all(info->cond);   
        cbus_mutex_unlock(info->lock);     
    }
   
    return G_OK;
}

static GErrc msg_response_add_info_async(cbus_client_t client,  cbus_client_response_t respone, cbus_message_id message_id, cbus_session_id session_id, cbus_client_on_response_func on_response, GPTR opaque)
{
    cbus_mutex_lock(client->lock);
    cbus_map_t session_map = NULL;

    if (cbus_map_find_pointer(respone, message_id, (const void**)&session_map) == GFALSE)
    {
        session_map = cbus_map_create(CMV_TYPE_Pointer, CBUS_MAP_DEFAULT_CAPACITY);

        if (!session_map)
        {
            cbus_mutex_unlock(client->lock);
            return G_ErrInsufficientResources;
        }  
    
        if (cbus_map_set_pointer(respone, message_id, session_map, client_map_free) == GFALSE)
        {
            cbus_map_destroy(session_map);
            cbus_mutex_unlock(client->lock);
            return G_ErrUndefined;
        }      
    }

    client_response_info_t* response_info = NULL;

    if (cbus_map_find_pointer(session_map, session_id, (const void**)&response_info) == GTRUE)
    {
        LOGE("session info for(msg id: %d, session id: %d) already exist, this is illegal\n", message_id, session_id);
        cbus_mutex_unlock(client->lock);
        return G_ErrAlreadyExist;
    }

    response_info = client_response_info_create(on_response, opaque);

    if (!response_info)
    {
        LOGE("create response info for(msg:%d, session:%d) failed\n", message_id, session_id);
        cbus_mutex_unlock(client->lock);
        return G_ErrInsufficientResources;
    }

    if (cbus_map_set_pointer(session_map, session_id, response_info, client_response_info_destroy) == GFALSE)
    {
        client_response_info_destroy(response_info);
        cbus_mutex_unlock(client->lock);
        return G_ErrUndefined;
    }

    cbus_mutex_unlock(client->lock);

    return G_OK;
}

static GErrc msg_response_add_info(cbus_client_t client,  cbus_client_response_t respone, cbus_message_id message_id, cbus_session_id session_id)
{
    return msg_response_add_info_async(client, respone, message_id, session_id, NULL, NULL);
}

static GErrc msg_response_wait(cbus_client_t client, cbus_client_response_t response, cbus_message_id message_id, cbus_session_id session_id, cbus_message_t* response_message, GU32 timeout_ms)
{
    client_response_info_t* response_info = NULL;
    cbus_map_t session_map = NULL;
    GU08 return_code = CBUS_MSG_OK;

    cbus_mutex_lock(client->lock);
    if (cbus_map_find_pointer(response, message_id, (const void**)&session_map) == GFALSE)
    {
        cbus_mutex_unlock(client->lock);
        return G_ErrNotFound;
    }

    if (cbus_map_find_pointer(session_map, session_id, (const void**)&response_info) == GFALSE)
    {
        cbus_mutex_unlock(client->lock);
        return G_ErrNotFound;
    }


    cbus_mutex_unlock(client->lock); /** todo, refinem e?? */

    cbus_mutex_lock(response_info->lock);
    if (timeout_ms == CBUS_CLIENT_WAIT_FOREVER)
    {
        while (response_info->response_message == NULL)
        {
            cbus_cv_wait(response_info->cond, response_info->lock);
        }
    }
    else
    {
        while (response_info->response_message == NULL)
        {
            GErrc ret = cbus_cv_wait_for(response_info->cond, response_info->lock, timeout_ms);
            if (ret != G_OK)
            {
                cbus_mutex_unlock(response_info->lock);
                return ret;
            }
        }        
    }

    return_code = cbus_message_get_return_code(response_info->response_message);

    if (response_message)
    {
        *response_message = response_info->response_message;
        response_info->response_message = NULL;
    }

    cbus_mutex_unlock(response_info->lock);

    cbus_mutex_lock(client->lock);
    cbus_map_remove(session_map, session_id);
    cbus_mutex_unlock(client->lock);

    return cbus_msg_return_code_to_gerrc((cbus_message_return_code_t)return_code);
}


static cbus_client_response_t get_response_handle(cbus_client_t client, GU08 message_type)
{    
    return client->response_map[message_type];    
}


static GErrc client_message_request(cbus_client_t client, cbus_message_t request_message, cbus_message_t* response_message, GU32  timeout_ms)
{
    if (!client || !request_message)
    {
        return G_ErrBadParameter;        
    }

    cbus_client_response_t response_handle = get_response_handle(client, cbus_message_get_response_type(cbus_message_get_message_type(request_message)));

    if (!response_handle)
    {
        return G_ErrInvalidStatus;
    }

    cbus_message_set_session_id(request_message, cbus_client_get_session_id(client));

    if (timeout_ms != CBUS_CLIENT_NO_WAIT)
    {
        /** add response info to 'reseponse map' */
        GErrc ret = msg_response_add_info(client, response_handle, cbus_message_get_message_id(request_message), cbus_message_get_session_id(request_message));

        if (G_OK != ret)
        {
            return ret;
        }
    }

    GErrc ret = cbus_endpoint_send(client->endpoint, request_message);

    if (G_OK == ret)
    {
        /** do wait  until receive response or error occur */
        if (timeout_ms != CBUS_CLIENT_NO_WAIT)
        {
        
            ret = msg_response_wait(client, response_handle, cbus_message_get_message_id(request_message), cbus_message_get_session_id(request_message), response_message, timeout_ms);
        }
    }

    return ret;    
}

static GErrc client_message_request2(cbus_client_t client, cbus_message_type_t message_type, cbus_message_id message_id, cbus_message_t* response_message, GU32  timeout_ms)
{
    cbus_message_t request_message = cbus_client_alloc_request_message(client, message_type, message_id, 0);

    if (!request_message)
    {
        return G_ErrInsufficientResources;
    }

    return client_message_request(client, request_message, response_message, timeout_ms);
}

static GErrc client_message_request_async(cbus_client_t client, cbus_message_t request_message,  cbus_client_on_response_func on_response, GPTR opaque)
{
    if (!client || !request_message)
    {
        return G_ErrBadParameter;        
    }

    cbus_client_response_t response_handle = get_response_handle(client, cbus_message_get_response_type(cbus_message_get_message_type(request_message)));

    if (!response_handle)
    {
        return G_ErrInvalidStatus;
    }


    GErrc ret = msg_response_add_info_async(client, response_handle, cbus_message_get_message_id(request_message), cbus_message_get_session_id(request_message), on_response, opaque);

    if (G_OK != ret)
    {
        return ret;
    }
    cbus_message_set_session_id(request_message, cbus_client_get_session_id(client));
    return cbus_endpoint_send(client->endpoint, request_message);    
}

static GErrc client_message_request_async2(cbus_client_t client, cbus_message_type_t message_type, cbus_message_id message_id,  cbus_client_on_response_func on_response, GPTR opaque)
{
    cbus_message_t request_message = cbus_client_alloc_request_message(client, message_type, message_id, 0);

    if (!request_message)
    {
        return G_ErrInsufficientResources;
    }

    return client_message_request_async(client, request_message, on_response, opaque);
}


static void client_on_method_response(cbus_message_t message)
{
    cbus_client_t client = (cbus_client_t)cbus_message_get_opaque(message);
    cbus_client_response_t response_handle = get_response_handle(client, cbus_message_get_message_type(message));
#if 0
    LOGI("[%s.%d]receive method respoonse, msg id: %d, type: %d, ssid: %d, response_handle: %p, opaque: %p\n", __FUNCTION__, __LINE__,
        cbus_message_get_message_id(message), cbus_message_get_message_type(message),
        cbus_message_get_session_id(message), response_handle, client);
#endif        
    if (response_handle)
    {
    
        msg_response_set_data(client, response_handle, message);
    }    
}

/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

static GErrc event_subscribe(cbus_map_t event_map, cbus_message_id message_id, cbus_client_on_event_func on_event, GPTR opaque)
{
    client_event_info_t* event_info = NULL;
    if (cbus_map_find_pointer(event_map, message_id, (const void**)&event_info) == GFALSE)
    {
        event_info = (client_event_info_t*)malloc(sizeof(client_event_info_t));
        if (!event_info)
        {
            return G_ErrInsufficientResources;
        }
        
        if (cbus_map_set_pointer(event_map, message_id, event_info, free) == GFALSE)
        {
            free(event_info);
            return G_ErrUndefined;
        }
    }

    event_info->on_event = on_event;
    event_info->opaque = opaque;
    return G_OK;
}

static GErrc event_stop_subscribe(cbus_map_t event_map, cbus_message_id message_id)
{
    return cbus_map_remove(event_map, message_id) ? G_OK : G_ErrNotFound;    
}

static client_event_info_t* event_get_info(cbus_map_t event_map, cbus_message_id message_id)
{
    client_event_info_t* event_info = NULL;
    cbus_map_find_pointer(event_map, message_id, (const void**)&event_info);

    return event_info;
}

/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
static void client_on_notification(cbus_client_t client, cbus_map_t event_map, cbus_message_t message)
{
    client_event_info_t* event_info = event_get_info(event_map, cbus_message_get_message_id(message));
    if (event_info && event_info->on_event)
    {
        cbus_message_set_opaque(message, event_info->opaque);
        event_info->on_event(message);
    }
}

static void cbus_client_on_event(cbus_message_t message)
{
    cbus_client_t client = (cbus_client_t)cbus_message_get_opaque(message);
#if 0
    LOGI("[%s.%d]receive method respoonse, msg id: %d, type: %d, ssid: %d\n", __FUNCTION__, __LINE__,
        cbus_message_get_message_id(message), cbus_message_get_message_type(message),
        cbus_message_get_session_id(message));
#endif
    switch (cbus_message_get_message_type(message))
    {
    case CBUS_MSG_TYPE_Event:                 client_on_notification(client, client->event_map, message);          break;
    case CBUS_MSG_TYPE_Property_Notification: client_on_notification(client, client->property_event_map, message); break;
    default:
        break;
    }
}

#define CLIENT_REG_MSG(endpoint_ref, msg_type, msg_func, opaque) do{ \
GErrc ret = cbus_endpoint_register_message_category(endpoint_ref, msg_type, msg_func, opaque); \
if (G_OK != ret){  \
    LOGE("reg msg category type: %d failed(0x%08x)\n", msg_type, ret); \
    return ret;    \
}}while(0)

static GErrc client_reg_message(cbus_client_t client)
{
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Method_Response,             client_on_method_response, client);
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Event_Subscribe_Ack,         client_on_method_response, client);
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Event_Stop_Subscribe_Ack,    client_on_method_response, client);
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Property_Get_Ack,            client_on_method_response, client);
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Property_Set_Ack,            client_on_method_response, client);
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Property_Subscribe_Ack,      client_on_method_response, client);
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Property_Stop_Subscribe_Ack, client_on_method_response, client);
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Event,                       cbus_client_on_event, client);
    CLIENT_REG_MSG(client->endpoint, CBUS_MSG_TYPE_Property_Notification,       cbus_client_on_event, client);

    return G_OK;
}

/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/

cbus_client_t cbus_client_create(const GSTR url, GU08 interface_version)
{
    if (!url)
    {
        return NULL;
    }

    cbus_client_t client = (cbus_client_t)malloc(sizeof(struct _cbus_client));

    if (!client)
    {
        return NULL;
    }

    memset(client, 0, sizeof(struct _cbus_client));
    client->lock = cbus_mutex_create();

    if (!client->lock)
    {
        cbus_client_destroy(client);
        return NULL;        
    }

    GErrc ret = cbus_endpoint_create(&client->endpoint, CBUS_EP_ROLE_CLIENT, NULL, url);

    if (G_OK != ret)
    {
        LOGE("create client endpoint failed(0x%08x)\n", ret);
        cbus_client_destroy(client);
        return NULL;
    }

    ret = response_map_init(client);
    if (G_OK != ret)
    {
        LOGE("initialize response map failed(0x%08x)\n", ret);
        cbus_client_destroy(client);
        return NULL;
    }

    client->event_map = cbus_map_create(CMV_TYPE_Pointer, CBUS_MAP_DEFAULT_CAPACITY);

    if (!client->event_map)
    {
        cbus_client_destroy(client);
        return NULL;
    }

    client->property_event_map = cbus_map_create(CMV_TYPE_Pointer, CBUS_MAP_DEFAULT_CAPACITY);

    if (!client->property_event_map)
    {
        cbus_client_destroy(client);
        return NULL;        
    }

    ret = client_reg_message(client);

    if (G_OK != ret)
    {
        LOGE("client register message to endpoint failed(0x%08x)\n", ret);
        cbus_client_destroy(client);
        return NULL;
    }

    ret = cbus_endpoint_start(client->endpoint);
    if (G_OK != ret)
    {
        LOGE("start client endpoint failed(0x%08x)\n", ret);
        cbus_client_destroy(client);
        return NULL;
    }

    return client;
}

void cbus_client_destroy(cbus_client_t client)
{
    if (client)
    {
        cbus_endpoint_destroy(client->endpoint);
        for (GU32 i = 0; i < CBUS_MSG_TYPE_MAX; i++)
        {
            if (client->response_map[i])
            {
                cbus_map_destroy(client->response_map[i]);
                client->response_map[i] = NULL;
            }
        }

        if (client->event_map)
        {
            cbus_map_destroy(client->event_map);
            client->event_map = NULL;
        }

        if (client->property_event_map)
        {
            cbus_map_destroy(client->property_event_map);
            client->property_event_map = NULL;
        }

        if (client->lock)
        {
            cbus_mutex_destroy(client->lock);
        }

        free(client);
    }
}

cbus_message_t cbus_client_alloc_request_message(cbus_client_t client, cbus_message_type_t message_type, cbus_message_id message_id, GU32 payload_size)
{
    if (!client)
    {
        return NULL;
    }

    cbus_message_t request_message = cbus_message_alloc(payload_size);
    if (request_message)
    {
        cbus_message_set_message_type(request_message, message_type);
        cbus_message_set_message_id(request_message, message_id);
        cbus_message_set_interface_version(request_message, client->interface_version);
        //cbus_message_set_session_id(request_message, cbus_client_get_session_id(client));
    }

    return request_message;
}


GErrc cbus_client_request(cbus_client_t client, cbus_message_t request_message, cbus_message_t* response_message, GU32 timeout_ms)
{
    if (!client)
    {
        return G_ErrBadParameter;
    }

    return client_message_request(client, request_message, response_message, timeout_ms);
}

GErrc cbus_client_request_async(cbus_client_t client, cbus_message_t request_message, cbus_client_on_response_func on_response, GPTR opaque)
{/** on_response == NULL, indicate that current don't care about the return response data */
    if (!client)
    {
        return G_ErrBadParameter;
    }

    return client_message_request_async(client, request_message, on_response, opaque);
}

GErrc cbus_client_request_no_return(cbus_client_t client, cbus_message_t request_message)
{
    if (!client)
    {
        return G_ErrBadParameter;
    }
    cbus_message_set_session_id(request_message, cbus_client_get_session_id(client));
    return cbus_endpoint_send(client->endpoint, request_message);
}


GErrc cbus_client_subscribe(cbus_client_t client, const cbus_client_event_t* event_info, GU32 timeout_ms)
{
    if (!client || !event_info || !event_info->on_event)
    {
        return G_ErrBadParameter;
    }

    GErrc ret = event_subscribe(client->event_map, event_info->event_id, event_info->on_event, event_info->opaque);

    if (G_OK != ret)
    {
        return ret;
    }

    //todo message register to message_manager
    
    return client_message_request2(client, CBUS_MSG_TYPE_Event_Subscribe, event_info->event_id, NULL, timeout_ms);

}

GErrc cbus_client_subscribe_group(cbus_client_t client, const cbus_client_event_t* group, GU32  group_size, GU32 timeout_ms)
{
    if (!client || !group || (group_size == 0))
    {
        return G_ErrBadParameter;
    }

    for (GU32 i = 0; i < group_size; i++)
    {
        GErrc ret = cbus_client_subscribe(client, &group[i], timeout_ms);

        if (G_OK != ret)
        {
            LOGE("subscribe event(%d) failed\n", group[i].event_id);
            return ret;
        }
    }
    
    return G_OK;
}

GErrc cbus_client_stop_subscribe(cbus_client_t client, cbus_message_id event_id, GU32 timeout_ms)
{
    if (!client)
    {
        return G_ErrBadParameter;
    }

    event_stop_subscribe(client->event_map, event_id);

    return client_message_request2(client, CBUS_MSG_TYPE_Event_Stop_Subscribe, event_id, NULL, timeout_ms);
}


GErrc cbus_client_property_get(cbus_client_t client, cbus_message_id property_id, cbus_message_t* response_message, GU32 timeout_ms)
{
    if (!client || !response_message)
    {
        return G_ErrBadParameter;
    }
    return client_message_request2(client, CBUS_MSG_TYPE_Property_Get, property_id, response_message, timeout_ms);
}

GErrc cbus_client_property_get_sync(cbus_client_t client, cbus_message_id property_id, cbus_client_on_response_func on_response, GPTR opaque)
{
    if (!client || !on_response)
    {
        return G_ErrBadParameter;
    }

    return client_message_request_async2(client, CBUS_MSG_TYPE_Property_Get, property_id, on_response, opaque);
}


GErrc cbus_client_property_set(cbus_client_t client, cbus_message_t request_message, cbus_message_t* response_message, GU32 timeout_ms)
{
    if (!client || !request_message) /** if response_message is NULL, indicate that here don't care about return value of current porperty */
    {
        return G_ErrBadParameter;
    }  
     
    return client_message_request(client, request_message, response_message, timeout_ms);
}

GErrc cbus_client_property_set_sync(cbus_client_t client, cbus_message_t request_message, cbus_client_on_response_func on_response, GPTR opaque)
{
    if (!client || !request_message) /** if on_response == NULL, indicate that, don't care about the return response data */
    {
        return G_ErrBadParameter;
    }

    return client_message_request_async(client, request_message, on_response, opaque);
}

GErrc cbus_client_property_subcscribe(cbus_client_t client, const cbus_client_property_t* property_info, GU32 timeout_ms)
{
    if (!client || !property_info || !property_info->on_event)
    {
        return G_ErrBadParameter;
    }

    GErrc ret = event_subscribe(client->property_event_map, property_info->property_id, property_info->on_event, property_info->opaque);

    if (G_OK != ret)
    {
        return ret;
    }

    return client_message_request2(client, CBUS_MSG_TYPE_Property_Subscribe, property_info->property_id, NULL, timeout_ms);
}

GErrc cbus_client_property_subcscribe_group(cbus_client_t client, const cbus_client_property_t* group, GU32 group_size, GU32 timeout_ms)
{
    if (!client || !group || (group_size == 0))
    {
        return G_ErrBadParameter;
    }

    for (GU32 i = 0; i < group_size; i++)
    {
        GErrc ret = cbus_client_property_subcscribe(client, &group[i], timeout_ms);

        if (G_OK != ret)
        {
            LOGE("subscribe event(%d) failed\n", group[i].property_id);
            return ret;
        }
    }
    
    return G_OK;
}

GErrc cbus_client_property_stop_subscribe(cbus_client_t client, cbus_message_id property_id, GU32 timeout_ms)
{
    if (!client)
    {
        return G_ErrBadParameter;
    }

    event_stop_subscribe(client->property_event_map, property_id);

    return client_message_request2(client, CBUS_MSG_TYPE_Property_Stop_Subscribe, property_id, NULL, timeout_ms);
}
